package team.bluepen.supermarket.data.hdfs;

import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import team.bluepen.supermarket.data.entity.HbaseItem;
import team.bluepen.supermarket.data.hbase.LineReader;

import java.io.IOException;

/**
 * @author Kai
 */
public abstract class TableReadMapper<T extends HbaseItem<T>> extends Mapper<LongWritable, Text, ImmutableBytesWritable, Put> {
    private final LineReader<T> reader;

    protected TableReadMapper(LineReader<T> reader) {
        this.reader = reader;
    }

    @Override
    protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, ImmutableBytesWritable, Put>.Context context) throws IOException, InterruptedException {
        String lineValue = value.toString();
        T t = reader.readLine(lineValue);
        ImmutableBytesWritable rowKeyWritable = new ImmutableBytesWritable(Bytes.toBytes(t.getRow()));
        Put put = reader.buildPut(t);
        context.write(rowKeyWritable, put);
    }
}